/*
 * TcpcRtpSocket.cpp
 *
 *  Created on: 2018年5月1日
 *      Author: zhengboyuan
 */

#include "TcpcRtpSocket.h"
#include "RtpHeader.h"
#include "RtpConst.h"
#include "RtpPort.h"
#include "CLog.h"

namespace av
{

TcpcRtpSocket::TcpcRtpSocket():
		m_state(kDisconnected),
		m_offset()
{
	m_buffer.ensure(RtpConst::MTU_MAX_SIZE);
	m_recvBuffer.ensure(RtpConst::MTU_MAX_SIZE * 2);
}

TcpcRtpSocket::~TcpcRtpSocket()
{
	close();
}

bool TcpcRtpSocket::open(const char* ip, int port)
{
	CLog::debug("TcpcRtpSocket::open. %s. %s:%d\n", CLog::getShortTime(), ip, port);

	if (!m_socket.open(SOCK_STREAM))
	{
		return false;
	}

	m_socket.setReuse();
	m_socket.setNonblock(true);

	m_socket.trySetSendBufferSize(SEND_BUFFER_SIZE, MIN_BUFFER_SIZE);
	m_socket.trySetRecvBufferSize(SEND_BUFFER_SIZE, MIN_BUFFER_SIZE);

	bool done = false;
	if (port == 0)
	{
		for (size_t i = 0; i < (RtpPort::count() + 1); ++ i)
		{
			port = RtpPort::make();

			comn::SockAddr addr(ip, port);
			if (0 == m_socket.bind(addr))
			{
				m_addr = addr;
				done = true;
				break;
			}
		}
	}
	else
	{
		comn::SockAddr addr(ip, port);
		if (0 == m_socket.bind(addr))
		{
			m_addr = addr;
			done = true;
		}
	}

	if (!done)
	{
		close();
		return false;
	}

	m_socketPair.open();

	return true;
}

void TcpcRtpSocket::close()
{
	CLog::debug("TcpcRtpSocket::close. %s\n", CLog::getShortTime());

	m_socket.close();

	setState(kDisconnected);

	m_socketPair.close();
}

bool TcpcRtpSocket::isOpen()
{
	return m_socket.isOpen();
}

NetAddress TcpcRtpSocket::getLocalAddr()
{
	return NetAddress(m_addr.getIPAddr(), m_addr.getPort());
}

bool TcpcRtpSocket::setPeer(const char* ip, int port)
{
	CLog::info("TcpcRtpSocket::setPeer %s:%d", ip, port);

	BasicRtpSocket::setPeer(ip, port);

	checkConnect(0);
	return true;
}

bool TcpcRtpSocket::send(RtpPacket& packet, int ms)
{
	//CLog::warning("TcpcRtpSocket::send. %s. \n", CLog::getShortTime());

	if (!checkConnect(ms))
	{
		CLog::warning("TcpcRtpSocket::send. %s. checkConnect failed. \n", CLog::getShortTime());
		return false;
	}

	if (!flushWrite(ms))
	{
		CLog::warning("TcpcRtpSocket::send. %s. failed flush.\n", CLog::getShortTime());
		return false;
	}

	bool done = false;

	m_offset = 0;

	m_buffer.ensure(packet.size + sizeof(RTP_FIXED_HEADER));

	RTP_FIXED_HEADER* header = (RTP_FIXED_HEADER*)m_buffer.data();
	header->version = 2;
	header->marker = packet.mark;

	header->timestamp = htonl(m_initTime + packet.ts);

	header->ssrc = htonl(m_ssrc);
	header->seq_no = htons(m_seq ++);
	header->payload = packet.pt;

	uint8_t* data = m_buffer.data() + sizeof(RTP_FIXED_HEADER);
	memcpy(data, packet.data, packet.size);

	int size = packet.size + sizeof(RTP_FIXED_HEADER);
	m_buffer.resize(size);
	data = m_buffer.data();

	int ret = send(data, size, ms);
	if (ret > 0)
	{
		if (ret == size)
		{
			m_buffer.clear();
			done = true;
		}
		else
		{
			m_offset = ret;
		}
	}

	//CLog::warning("TcpcRtpSocket::send. %s. return %d\n", CLog::getShortTime(), done);

	return done;
}

bool TcpcRtpSocket::recv(RtpPacketHandler handler, int ms)
{
	if (!checkConnect(ms))
	{
		return false;
	}

	//if (!m_socket.checkReadable(ms))
	if (m_socketPair.checkReadable(m_socket.getHandle(), ms))
	{
		return false;
	}

	uint8_t* data = m_recvBuffer.data();
	int size = m_recvBuffer.size();
	uint8_t* buf = data + size;

	int hdrSize = sizeof(int32_t) + sizeof(RTP_FIXED_HEADER);
	if (size < hdrSize)
	{
		int32_t toRead = hdrSize - size;
		int ret = m_socket.receive((char*)buf, toRead, 0);
		if (ret == 0)
		{
			setState(kDisconnected);
			m_recvBuffer.clear();
			return false;
		}
		else if (ret < 0)
		{
			return false;
		}
		else
		{
			m_recvBuffer.resize(size + ret);
			if (ret == toRead)
			{
				int32_t length = *((int32_t*)data);
				length = ntohl(length);

				m_recvBuffer.ensure(length);

				data = m_recvBuffer.data();
				size = m_recvBuffer.size();
				buf = data + size;
			}
			else
			{
				return false;
			}
		}
	}

	int32_t length = *((int32_t*)data);
	length = ntohl(length);

	int toRead = length - 4 - size;
	int ret = m_socket.receive((char*)buf, toRead, 0);
	if (ret == 0)
	{
		setState(kDisconnected);
		m_recvBuffer.clear();
		return false;
	}
	else if (ret < 0)
	{
		return false;
	}
	else
	{
		m_recvBuffer.resize(size + ret);

		if (ret == toRead)
		{
			data = m_recvBuffer.data() + 4;
			size = length;

			RTP_FIXED_HEADER* header = (RTP_FIXED_HEADER*)data;

			data += sizeof(RTP_FIXED_HEADER);
			size -= sizeof(RTP_FIXED_HEADER);

			RtpPacket pkt;
			memset(&pkt, 0, sizeof(pkt));
			pkt.pt = header->payload;
			pkt.mark = header->marker;
			pkt.ts = header->timestamp;
			pkt.data = data;
			pkt.size = size;

			handler(pkt);

			m_recvBuffer.clear();

			return true;
		}
		else
		{
			return false;
		}
	}
}

bool TcpcRtpSocket::isConnected()
{
	comn::AutoCritSec lock(m_cs);
	return isOpen() && (m_state == kConnected);
}

bool TcpcRtpSocket::checkConnect(int ms)
{
	if (m_state == kConnected)
	{
		return true;
	}

	if (m_state == kDisconnected)
	{
		m_socket.connect(m_peerAddr);

		setState(kConnecting);
		fireConnectionState(kConnecting);
	}

	if (m_socketPair.checkWritable(m_socket.getHandle(), ms))
	//if (m_socket.checkWritable(ms))
	{
		setState(kConnected);
		fireConnectionState(kConnected);
	}
	else
	{
		setState(kDisconnected);
		fireConnectionState(kDisconnected);
	}

	return (m_state == kConnected);
}

bool TcpcRtpSocket::flushWrite(int ms)
{
	if (m_offset < 0 || (m_offset >= (int)m_buffer.size()))
	{
		return true;
	}

	uint8_t* data = m_buffer.data() + m_offset;
	int size = m_buffer.size() - m_offset;

	int ret = send(data, size, ms);
	if (ret > 0)
	{
		m_offset += ret;
	}
	return (m_offset >= (int)m_buffer.size());
}

/// 发送并检测断开
int TcpcRtpSocket::send(uint8_t* data, int size, int ms)
{
	int ret = m_socket.sendBlock((char*)data, size);
	if (ret == -1)
	{
		int error = comn::Socket::getLastError();
		if (error == ECONNRESET)
		{
			setState(kDisconnected);
			fireConnectionState(kDisconnected);
		}
	}
	return ret;
}

bool TcpcRtpSocket::checkConnected()
{
	return true; //todo
}

void TcpcRtpSocket::setState(State state)
{
	comn::AutoCritSec lock(m_cs);
	m_state = state;
}

RtpSocket::State TcpcRtpSocket::getState()
{
	comn::AutoCritSec lock(m_cs);
	return m_state;
}

void TcpcRtpSocket::abort()
{
    m_socketPair.makeReadable();
}



} /* namespace av */
